package brokerFailureDetector;

import java.util.HashMap;
import java.util.Set;

import org.apache.qpid.client.TopicPublisherAdapter;

import qpidClient.QpidClient;

/**
 * Gets the brokers which should be working and creates threads to check if
 * each one of them is working well.
 * Sends a message to the controller when the working state (=health) of a
 * broker changes.
 */
public class Detector extends QpidClient {
	
    //machine in which the detector is running
    protected final String HOST_IP;
    
	//list of machines in which a broker is running and their state
    //state is an int from 0 (worst) to 10 (best), null if we don't know
    protected HashMap<String, Integer> brokersState = new HashMap<String, Integer>();
    
    //machine in which the controller is running
    protected final String CONTROLLER_IP;
    //topic used to communicate with the controller
    protected final String TOPIC_CONTROLLER = "brokerFailure.controller";
    //publisher which will send messages to the controller
    TopicPublisherAdapter publisherToController;
    
    //getters
	public String getHOST_IP() {
		return HOST_IP;
	}

	/**
	 * Constructor.
	 * 
	 * @param hostIP  IP address of the machine the detector is running on
	 * @param controllerIP  IP address of the controller
	 * @param brokerAddresses  list of machines in which a broker is running
	 * 						   and the port in which they are listening
	 */
    public Detector(String hostIP, String controllerIP,
    				HashMap<String, Integer> brokerAddresses) {
    	
    	super(); //assign ID
    	
    	HOST_IP = hostIP;
    	CONTROLLER_IP = controllerIP;
 	   	
 	   	//start a connection to send messages to the controller
 	   	try {
 	   		//create and start a connection and a session in it
			openConnAndSess(CONTROLLER_IP);
			//create publisher
			publisherToController = getPublisherForTopic(TOPIC_CONTROLLER);
 	   	} catch (Exception e) {
			e.printStackTrace();
 	   	}
 	   	
 	   Set<String> brokers = brokerAddresses.keySet();
   	
 	   //for each broker
 	   for (String broker : brokers) {
 		   //put its initial state (off) in brokersState
 		   brokersState.put(broker, 0);
 		   //create a thread
 		   new DetectorThread(this, broker, brokerAddresses.get(broker));
		}
 	   	
 	   System.out.println("Detector in " + HOST_IP + " ready");
   	
 	   while (true) {
			//wait to see if any thread detects any failure in any broker
		}
        
    }
    
    /**
     * Receives the most recent state of a broker. If it is not the same as the
     * previous state of that broker, informs the controller about the change.
     * 
     * @param broker  whose state is being updated
     * @param state  new state of the broker
     */
    public void updateState(String broker, int state) {
    	//get the previous state
    	int oldState = brokersState.get(broker);
    	//save the new state
    	brokersState.put(broker, state);
    	//decide when to report a change to the controller
    	if (oldState != state) {
//    		System.out.println("HEY, THE BROKER IN " + HOST_IP
//    						   + " DETECTED THAT THE BROKER IN " + broker
//    						   + " IS CHANGING ITS STATE FROM " + oldState
//    						   + " TO " + state);
    		//send message to broker
    		//create the text of the message to send
    		String text = HOST_IP + ":" + broker + ":" + state;
			try {
				//send the message to the controller
				sendMessage(publisherToController, text);
			} catch (Exception e) {
				e.printStackTrace();
			}
    	}
//    	System.out.println("State in " + HOST_BROKER + ": " + brokersState);
    }

}
